package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

//订单预处理表  卢顺祥
public class DwdTradeOrderPreProcess {
    public static void main(String[] args) {
//TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15*60 + 10));
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的topic_db主题中读取数据 创建动态表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_order_pre_group"));

        //TODO 4.过滤出订单明细
        Table orderDetail = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['order_id'] order_id,\n" +
                "data['course_id'] course_id,\n" +
                "data['course_name'] course_name,\n" +
                "data['create_time'] create_time,\n" +
                //原始金额
                "data['origin_amount'] split_original_amount,\n" +
                //减免金额
                "data['coupon_reduce'] split_coupon_amount,\n" +
                //最终金额
                "data['final_amount'] split_total_amount,\n" +
                "ts od_ts,\n" +
                "proc_time\n" +
                "from `topic_db` where `table` = 'order_detail' " +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail", orderDetail);
        //tableEnv.executeSql("select * from order_detail").print();


        //TODO 5.过滤出订单
        Table orderInfo = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['province_id'] province_id,\n" +
                "data['create_time'] create_time,\n" +
                "data['order_status'] order_status,\n" +
                "`type`,\n" +
                "ts oi_ts\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_info'\n" +
                "and (`type` = 'insert' or `type` = 'update')");
        tableEnv.createTemporaryView("order_info", orderInfo);

        //TODO 9.将上述5张表进行关联
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "od.id,\n" +
                "od.order_id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "date_format(od.create_time, 'yyyy-MM-dd') date_id,\n" +
                "od.create_time,\n" +
                "od.split_original_amount,\n" +
                "od.split_coupon_amount,\n" +
                "od.split_total_amount,\n" +
                "od.od_ts,\n" +

                "oi.user_id,\n" +
                "oi.province_id,\n" +
                "date_format(oi.create_time, 'yyyy-MM-dd') operate_date_id,\n" +
                "oi.order_status,\n" +
                "oi.`type`,\n" +
                "oi.oi_ts,\n" +

                "current_row_timestamp() row_op_ts\n" +
                "from order_detail od \n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id");
        tableEnv.createTemporaryView("result_table", resultTable);

        //TODO 10.创建动态表 和要写入的kafka主题进行映射

        tableEnv.executeSql("create table dwd_trade_order_pre_process(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "split_original_amount string,\n" +
                "split_coupon_amount string,\n" +
                "split_total_amount string,\n" +
                "od_ts string,\n" +

                "user_id string,\n" +
                "province_id string,\n" +
                "operate_date_id string,\n" +
                "order_status string,\n" +
                "`type` string,\n" +
                "oi_ts string,\n" +

                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));

       //TODO 11.写入
        tableEnv.executeSql("insert into dwd_trade_order_pre_process select * from result_table");

    }
}
